package org.tao.lightningmq.broker.core;

import com.alibaba.fastjson2.JSON;
import io.netty.channel.ChannelHandlerContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tao.lightningmq.broker.cache.CommonCache;
import org.tao.lightningmq.broker.enums.BrokerClusterModeEnum;
import org.tao.lightningmq.common.constants.BrokerConstants;
import org.tao.lightningmq.common.dto.MessageDTO;
import org.tao.lightningmq.common.dto.SendMessageToBrokerResponseDTO;
import org.tao.lightningmq.common.dto.SlaveSyncRespDTO;
import org.tao.lightningmq.common.enums.*;
import org.tao.lightningmq.common.event.model.BaseEvent;
import org.tao.lightningmq.common.remote.SyncFuture;
import org.tao.lightningmq.common.remote.SyncFutureManager;
import org.tao.lightningmq.common.tcp.TcpMsg;

import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

/**
 * @Author lht
 * @date 2025/1/3 - 16:30
 * @description:
 */
public class CommitLogAppendHandler {

    private static final Logger LOGGER = LoggerFactory.getLogger(CommitLogAppendHandler.class);

    public void prepareMMapLoading(String topicName) throws IOException {
        CommitLogMMapFileModel commitLogMMapFileModel = new CommitLogMMapFileModel();
        commitLogMMapFileModel.loadFileInMMap(topicName, 0, BrokerConstants.COMMIT_LOG_DEFAULT_MMAP_SIZE, FileSizeUnitEnum.B);
        CommonCache.getCommitLogMMapFileModelManager().put(topicName, commitLogMMapFileModel);
    }

    public void appendMsg(MessageDTO messageDTO) throws IOException {
        CommitLogMMapFileModel commitLogMMapFileModel = CommonCache.getCommitLogMMapFileModelManager().get(messageDTO.getTopic());
        if (commitLogMMapFileModel == null) {
            throw new RuntimeException("topic: " + messageDTO.getTopic() + " not found");
        }
        commitLogMMapFileModel.writeContent(messageDTO, true);
    }

    public void appendMsg(MessageDTO messageDTO, BaseEvent event) throws Exception {
        appendMsg(messageDTO);
        int sendWay = messageDTO.getSendWay();
        boolean isAsyncSend = MessageSendWay.ASYNC.getCode() == sendWay;
        //判断下是主节点还是从节点
        boolean isClusterMode = BrokerClusterModeEnum.MASTER_SLAVE.getCode().equals(CommonCache.getGlobalProperties().getBrokerClusterMode());
        boolean isMasterNode = "master".equals(CommonCache.getGlobalProperties().getBrokerClusterRole());
        boolean isDelayMsg = messageDTO.getDelay() > 0;
        if (isClusterMode) {
            if (isMasterNode) {
                SendMessageToBrokerResponseDTO sendMsgResp = new SendMessageToBrokerResponseDTO();
                sendMsgResp.setMsgId(messageDTO.getMsgId());
                List<SyncFuture> needAckTask = new LinkedList<>();
                List<String> needRemoveMsgIdList = new LinkedList<>();
                // 需要同步给从节点
                for (ChannelHandlerContext slaveChannel : CommonCache.getSlaveChannelMap().values()) {
                    if (!slaveChannel.channel().isActive()) {
                        LOGGER.warn("slave channel is not active, channel: {}", slaveChannel);
                        continue;
                    }
                    String newMsgId = UUID.randomUUID().toString();
                    messageDTO.setMsgId(newMsgId);
                    SyncFuture syncFuture = new SyncFuture();
                    syncFuture.setMsgId(newMsgId);
                    needAckTask.add(syncFuture);
                    needRemoveMsgIdList.add(newMsgId);
                    SyncFutureManager.put(newMsgId, syncFuture);
                    slaveChannel.writeAndFlush(new TcpMsg(BrokerEventCode.PUSH_MSG.getCode(), JSON.toJSONBytes(messageDTO)));
                }
                if (isAsyncSend || isDelayMsg) {
                    return;
                }
                // 同步复制模式 等待从节点ack
                int cur = 0;
                while (!needAckTask.isEmpty() && cur++ < 10) {
                    needAckTask.removeIf(SyncFuture::isDone);
                    if (!needAckTask.isEmpty())
                        TimeUnit.MICROSECONDS.sleep(50);
                }
                // 从节点ack超时
                if (!needAckTask.isEmpty()) {
                    sendMsgResp.setStatus(SendMessageToBrokerResponseStatus.FAIL.getCode());
                    sendMsgResp.setDesc("Slave node sync fail! Sync task had TimeoutException");
                    LOGGER.warn("wait slave ack timeout, msgId: {}", needAckTask.get(0).getMsgId());
                    // FIXME 从节点ACK失败后
                } else {
                    // 从节点ack成功
                    sendMsgResp.setStatus(SendMessageToBrokerResponseStatus.SUCCESS.getCode());
                }
                TcpMsg responseMsg = new TcpMsg(BrokerResponseCode.SEND_MSG_RESP.getCode(), JSON.toJSONBytes(sendMsgResp));
                event.getContext().writeAndFlush(responseMsg);
                // 移除ack防止内存泄漏
                needRemoveMsgIdList.forEach(SyncFutureManager::remove);
            } else {
                if (isAsyncSend || isDelayMsg) {
                    return;
                }
                // 从节点，返回给主节点
                SlaveSyncRespDTO slaveSyncRespDTO = new SlaveSyncRespDTO();
                slaveSyncRespDTO.setMsgId(messageDTO.getMsgId());
                slaveSyncRespDTO.setSyncSuccess(true);
                event.getContext().writeAndFlush(new TcpMsg(BrokerResponseCode.SLAVE_SYNC_RESP.getCode(),
                        JSON.toJSONBytes(slaveSyncRespDTO)));
            }
        } else {
            if (isAsyncSend || isDelayMsg) {
                return;
            }
            SendMessageToBrokerResponseDTO resp = new SendMessageToBrokerResponseDTO();
            resp.setStatus(SendMessageToBrokerResponseStatus.SUCCESS.getCode());
            resp.setMsgId(messageDTO.getMsgId());
            TcpMsg tcpMsg = new TcpMsg(BrokerResponseCode.SEND_MSG_RESP.getCode(), JSON.toJSONBytes(resp));
            event.getContext().writeAndFlush(tcpMsg);
        }
    }

}
